package com.hibase.core.mq.redis.handler;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.hibase.common.constant.RedisKeyConstant;
import com.hibase.core.mq.annotation.HibaseQueue;
import com.hibase.core.mq.annotation.HibaseQueueProvider;
import com.hibase.core.mq.factory.QueueHandler;
import com.hibase.core.redis.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.lang.reflect.Method;
import java.util.concurrent.Executors;

/**
 * Redis 对 消息队列的实现
 */
@Slf4j
public class RedisQueueHandler implements QueueHandler {

    @Autowired
    private RedisUtil redisUtil;

    @Override
    public Object execute(Method method, Object[] args) throws Exception{
        long rId = 0;
        if(method.getDeclaringClass().isAnnotationPresent(HibaseQueue.class) && method.isAnnotationPresent(HibaseQueueProvider.class)){
            if(args == null || args.length != 1){
                throw new Exception("Redis Queue Provider Java Interface Defined Error.");
            }
            HibaseQueueProvider redisQueueProvider = method.getDeclaredAnnotation(HibaseQueueProvider.class);
            String queueName = StrUtil.isNotBlank(redisQueueProvider.value()) ? redisQueueProvider.value() : redisQueueProvider.queueName();
            if(StrUtil.isBlank(queueName)){
                throw new Exception("Redis Queue Provider Java Config Error,please Check your configuration");
            }
            switch (redisQueueProvider.strategy()){
                case SYNC:
                    try{
                        //左进
                        rId = redisUtil.lLeftPush(RedisKeyConstant.QUEUE_PREFIX_KEY+queueName, JSON.toJSONString(args[0]));
                        log.info("Sync Send Message [{}] To Queue [{}].", JSON.toJSONString(args[0]),queueName);
                    }catch (Exception e){
                        log.error("Sync Send Message [" + JSON.toJSONString(args[0]) + "] To Queue [" + queueName + "] Error.",e);
                    }
                    return rId;
                case ASYNC:
                    Executors.newSingleThreadExecutor().submit(()-> {
                        try{
                            redisUtil.lLeftPush(RedisKeyConstant.QUEUE_PREFIX_KEY+queueName, JSON.toJSONString(args[0]));
                            log.info("Async Send Message [{}] To Queue [{}].", JSON.toJSONString(args[0]),queueName);
                        }catch (Exception e){
                            log.error("ASync Send Message [" + JSON.toJSONString(args[0]) + "] To Queue [" + queueName + "] Error.",e);
                        }
                    });
                    return 1;
            }
        }
        return rId;
    }
}
